agentmux_srv\backend\blockcontroller/
core.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Shared helpers used by all block controllers (persistent, subprocess, acp).
5//!
6//! Extracted to avoid copy-paste across the four controller files. Each
7//! controller keeps only its transport-specific logic (PTY / stream-json / ACP)
8//! and delegates the common mechanics here.
9
10use std::collections::HashMap;
11use std::sync::Arc;
12
13use super::health::HealthMonitor;
14use crate::backend::eventbus::EventBus;
15use crate::backend::storage::store::Store;
16
17/// Block metadata key for the persisted agent session ID.
18/// Used by the "My Agents" reattach path on the frontend: after a tab reload
19/// the picker reads `block.meta["agent:sessionid"]` and passes it as the
20/// `--resume` value so the CLI picks up the prior conversation.
21pub(crate) const META_SESSION_ID: &str = "agent:sessionid";
22
23/// Block metadata key for the last classified agent failure.
24/// Written on every non-zero / in-band-error exit; cleared on clean success.
25/// The frontend reads this on pane mount so the recovery banner survives
26/// tab switches and page reloads without requiring the WPS event to be
27/// received in real time.
28pub(crate) const META_LAST_FAILURE: &str = "agent:last_failure";
29
30/// Expand the working directory, create it if missing, set it on `cmd`, and
31/// apply all env-var overrides from `env_vars`.
32///
33/// Call this from each controller's spawn routine BEFORE any transport-specific
34/// `Command` flags (e.g. `CREATE_NO_WINDOW` on Windows, which must come after
35/// the env setup in some implementations).
36pub(crate) fn apply_working_dir(
37    cmd: &mut tokio::process::Command,
38    block_id: &str,
39    working_dir: &str,
40    env_vars: &HashMap<String, String>,
41) {
42    if !working_dir.is_empty() {
43        let expanded_dir = expand_home_dir(working_dir);
44        let dir_path = std::path::Path::new(&expanded_dir);
45        if !dir_path.exists() {
46            if let Err(e) = std::fs::create_dir_all(dir_path) {
47                tracing::warn!(
48                    block_id = %block_id,
49                    dir = %expanded_dir,
50                    error = %e,
51                    "failed to create working directory",
52                );
53            }
54        }
55        if dir_path.exists() {
56            cmd.current_dir(&expanded_dir);
57            // Warn loudly if the agent workspace contains a nested .git (e.g.
58            // an unintended `git clone` inside ~/.agentmux/agents/). A stale
59            // nested clone can confuse agents into reading old code and waste
60            // gigabytes of disk. Single fs::metadata call — no directory walk.
61            let looks_like_agent_workspace = expanded_dir.contains("/.agentmux/agents/")
62                || expanded_dir.contains("\\.agentmux\\agents\\");
63            if looks_like_agent_workspace {
64                let git_dir = dir_path.join(".git");
65                if git_dir.exists() {
66                    tracing::warn!(
67                        block_id = %block_id,
68                        cwd = %expanded_dir,
69                        ".git detected inside agent workspace — this is usually \
70                         an unintended nested clone and can waste gigabytes of \
71                         disk. Clean up with: rm -rf {}/.git",
72                        expanded_dir,
73                    );
74                }
75            }
76        }
77    }
78    for (k, v) in env_vars {
79        let expanded = crate::backend::base::expand_home_dir_safe(v);
80        cmd.env(k, expanded.to_string_lossy().as_ref());
81    }
82}
83
84/// Expand a leading `~/` or bare `~` to the user's home directory.
85pub(crate) fn expand_home_dir(dir: &str) -> String {
86    if dir.starts_with("~/") || dir == "~" {
87        if let Some(home) = dirs::home_dir() {
88            return home
89                .join(dir.trim_start_matches("~/"))
90                .to_string_lossy()
91                .to_string();
92        }
93    }
94    dir.to_string()
95}
96
97/// Spawn the health-watchdog background task for a turn.
98///
99/// The watchdog polls `health_monitor.check()` every 5 s while a turn is
100/// active and exits as soon as `is_active_turn()` returns false (i.e. after
101/// the turn ends or the process exits). Duplicated verbatim in
102/// persistent.rs and subprocess.rs (twice) before this extraction.
103pub(crate) fn spawn_health_watchdog(health_monitor: &Arc<HealthMonitor>) {
104    let health = Arc::clone(health_monitor);
105    tokio::spawn(async move {
106        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(5));
107        loop {
108            interval.tick().await;
109            if !health.is_active_turn() {
110                break;
111            }
112            health.check();
113        }
114    });
115}
116
117/// Persist a newly captured session ID to block metadata and broadcast the
118/// `waveobj:update` event so the frontend reflects the change immediately.
119///
120/// This is the "careful" path from persistent.rs and subprocess.rs. ACP was
121/// missing this step (it only set `inner.session_id` in memory) — A5 fixes it
122/// by routing ACP through this same function.
123///
124/// No-ops silently when `wstore` is `None` (e.g. in unit tests that don't wire
125/// up a store).
126pub(crate) fn persist_session_id(
127    block_id: &str,
128    sid: &str,
129    wstore: &Option<Arc<Store>>,
130    event_bus: &Option<Arc<EventBus>>,
131) {
132    let Some(ref store) = wstore else {
133        return;
134    };
135    let oref_str = format!("block:{}", block_id);
136    let mut meta_update = crate::backend::obj::MetaMapType::new();
137    meta_update.insert(
138        META_SESSION_ID.to_string(),
139        serde_json::Value::String(sid.to_string()),
140    );
141    match crate::server::service::update_object_meta(store, &oref_str, &meta_update) {
142        Err(e) => {
143            tracing::warn!(
144                block_id = %block_id,
145                error = %e,
146                "failed to persist agent:sessionid",
147            );
148        }
149        Ok(_) => {
150            let Some(ref event_bus) = event_bus else {
151                return;
152            };
153            if let Ok(updated_block) =
154                store.must_get::<crate::backend::obj::Block>(block_id)
155            {
156                let update_data = serde_json::to_value(
157                    &crate::backend::obj::WaveObjUpdate {
158                        updatetype: "update".into(),
159                        otype: "block".into(),
160                        oid: block_id.to_string(),
161                        obj: Some(crate::backend::obj::wave_obj_to_value(&updated_block)),
162                    },
163                )
164                .ok();
165                event_bus.broadcast_event(&crate::backend::eventbus::WSEventType {
166                    eventtype: "waveobj:update".to_string(),
167                    oref: oref_str,
168                    data: update_data,
169                });
170            }
171        }
172    }
173}
174
175/// Persist or clear the last agent failure in block metadata.
176///
177/// Pass `Some(failure)` on a failed exit to write `agent:last_failure` into the
178/// block's meta so the pane can recover the recovery banner on any future load
179/// without needing the ephemeral WPS event. Pass `None` on a clean exit to
180/// remove the key (setting it to JSON null triggers `merge_meta`'s delete path).
181/// Broadcasts a `waveobj:update` so active frontend subscribers see the change
182/// immediately via the block atom, not just on next full load.
183pub(crate) fn persist_last_failure(
184    block_id: &str,
185    failure: Option<&crate::agents::failure::AgentFailure>,
186    wstore: &Option<Arc<Store>>,
187    event_bus: &Option<Arc<EventBus>>,
188) {
189    let Some(ref store) = wstore else {
190        return;
191    };
192    // On a clean exit (failure=None), only write null (which merge_meta uses to
193    // delete the key) if the key actually exists — otherwise every successful
194    // turn would trigger a redundant DB write + waveobj:update broadcast.
195    if failure.is_none() {
196        let key_exists = store
197            .must_get::<crate::backend::obj::Block>(block_id)
198            .ok()
199            .and_then(|b| b.meta.get(META_LAST_FAILURE).cloned())
200            .map(|v| !v.is_null())
201            .unwrap_or(false);
202        if !key_exists {
203            return;
204        }
205    }
206    let oref_str = format!("block:{}", block_id);
207    let mut meta_update = crate::backend::obj::MetaMapType::new();
208    let val = match failure {
209        Some(f) => match serde_json::to_value(f) {
210            Ok(v) => v,
211            Err(e) => {
212                tracing::warn!(
213                    block_id = %block_id,
214                    error = %e,
215                    "failed to serialize AgentFailure for agent:last_failure — skipping meta write",
216                );
217                return;
218            }
219        },
220        None => serde_json::Value::Null, // null → merge_meta removes the key
221    };
222    meta_update.insert(META_LAST_FAILURE.to_string(), val);
223    match crate::server::service::update_object_meta(store, &oref_str, &meta_update) {
224        Err(e) => {
225            tracing::warn!(
226                block_id = %block_id,
227                error = %e,
228                "failed to persist agent:last_failure",
229            );
230        }
231        Ok(_) => {
232            let Some(ref bus) = event_bus else {
233                return;
234            };
235            if let Ok(updated_block) =
236                store.must_get::<crate::backend::obj::Block>(block_id)
237            {
238                let update_data = serde_json::to_value(
239                    &crate::backend::obj::WaveObjUpdate {
240                        updatetype: "update".into(),
241                        otype: "block".into(),
242                        oid: block_id.to_string(),
243                        obj: Some(crate::backend::obj::wave_obj_to_value(&updated_block)),
244                    },
245                )
246                .ok();
247                bus.broadcast_event(&crate::backend::eventbus::WSEventType {
248                    eventtype: "waveobj:update".to_string(),
249                    oref: oref_str,
250                    data: update_data,
251                });
252            }
253        }
254    }
255}
agentmux_srv\backend\blockcontroller/core.rs

agentmux_srv\backend\blockcontroller/
core.rs
